package io.kiki.sba.registry.server.metadata.multi.cluster;

import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.elector.Leader;
import io.kiki.sba.registry.common.model.multi.cluster.DataCenter;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.exception.MetaLeaderNotWarmupException;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.metadata.bootstrap.config.MultiClusterMetaServerConfig;
import io.kiki.sba.registry.server.metadata.multi.cluster.remote.RemoteClusterCentralControlExchanger;
import io.kiki.sba.registry.server.metadata.multi.cluster.remote.RemoteClusterSlotSyncRequest;
import io.kiki.sba.registry.server.metadata.multi.cluster.remote.RemoteClusterSlotSyncResponse;
import io.kiki.sba.registry.task.KeyedTask;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.StringFormatter;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class MultiClusterSlotsSyncerImpl implements MultiClusterSlotsSyncer {

    static final int MAX_SYNC_FAIL_COUNT = 3;
    private static final Logger logger = LoggerFactory.getLogger(MultiClusterSlotsSyncerImpl.class);
    public static volatile long LAST_REFRESH_CONFIG_TS = 0;
    public Map<String, RemoteClusterSlotState> dataCenterToRemoteClusterSlotStateMap = new ConcurrentHashMap<>();
    public SlotMapWatcher slotMapWatcher = new SlotMapWatcher(this);
    @Autowired
    public MetadataLeaderService metadataLeaderService;
    @Autowired
    public MultiClusterMetaServerConfig multiClusterMetaServerConfig;
    @Autowired
    public RemoteClusterCentralControlExchanger remoteClusterMetaExchanger;
    @Getter
    @Autowired
    public ExecutorManager executorManager;

    @PostConstruct
    public void init() {
        remoteClusterMetaExchanger.refreshDataCenterToMultiClusterSyncInfoMap();
        ConcurrentUtils.createDaemonThread("multi_cluster_slot_table", slotMapWatcher).start();
        metadataLeaderService.registerListener(this);
    }

    @Override
    public void becomeLeader() {
        slotMapWatcher.wakeup();
    }

    @Override
    public void loseLeader() {
    }

    @Override
    public Map<String, RemoteClusterSlotState> getDataCenterToRemoteClusterSlotStateMap() {
        if (!metadataLeaderService.amIStableAsLeader()) {
            throw new MetaLeaderNotWarmupException(metadataLeaderService.getLeader(), metadataLeaderService.getLeaderEpoch());
        }

        Map<String, RemoteClusterSlotState> map = new HashMap<>(dataCenterToRemoteClusterSlotStateMap.size());
        for (Entry<String, RemoteClusterSlotState> entry : dataCenterToRemoteClusterSlotStateMap.entrySet()) {
            RemoteClusterSlotState remoteClusterSlotState = entry.getValue();
            DataCenter dataCenter = remoteClusterSlotState.getDataCenterMetadata();
            if (dataCenter == null) {
                logger.error("[getMultiClusterSlots]dataCenter: {} metadata is null.", entry.getKey());
                continue;
            }
            map.put(dataCenter.getDataCenter(), new RemoteClusterSlotState(remoteClusterSlotState.getSlots(), dataCenter));
        }
        return map;
    }


    public boolean needReloadConfig() {
        return System.currentTimeMillis() - LAST_REFRESH_CONFIG_TS > multiClusterMetaServerConfig.getMultiClusterConfigReloadMillis();
    }

    /**
     * need sync slot table from remote cluster
     */
    public boolean needSync(KeyedTask<RemoteClusterSlotSyncTask> task) {
        return task == null || task.isOverAfter(multiClusterMetaServerConfig.getRemoteSlotSyncerMillis());
    }

    public void handleSyncResponse(RemoteClusterSlotSyncRequest remoteClusterSlotSyncRequest, Object response) {
        RemoteClusterSlotState remoteClusterSlotState = dataCenterToRemoteClusterSlotStateMap.get(remoteClusterSlotSyncRequest.getDataCenter());
        if (!(response instanceof GenericResponse)) {
            throw new RuntimeException(StringFormatter.format("sync request: {} fail, resp: {}", remoteClusterSlotSyncRequest, response));
        }
        GenericResponse<RemoteClusterSlotSyncResponse> remoteClusterSlotSyncResponseGenericResponse = (GenericResponse<RemoteClusterSlotSyncResponse>) response;
        RemoteClusterSlotSyncResponse remoteClusterSlotSyncResponse = remoteClusterSlotSyncResponseGenericResponse.getData();

        if (remoteClusterSlotSyncResponseGenericResponse.isSuccess()) {
            boolean learn = remoteClusterMetaExchanger.learn(remoteClusterSlotSyncRequest.getDataCenter(), new Leader(remoteClusterSlotSyncResponse.getMetaLeader(), remoteClusterSlotSyncResponse.getMetaLeaderEpoch()));
            if (learn) {
                handleSyncResult(remoteClusterSlotState, remoteClusterSlotSyncResponse);
                remoteClusterSlotState.initFailCount();
            }
            return;
        }
        handleFailResponse(remoteClusterSlotSyncRequest, remoteClusterSlotSyncResponseGenericResponse);
    }

    public void handleSyncResult(RemoteClusterSlotState remoteClusterSlotState, RemoteClusterSlotSyncResponse remoteClusterSlotSyncResponse) {

        if (remoteClusterSlotSyncResponse.isSlotsUpgrade()) {
            logger.info("Slots data upgrade:{}", remoteClusterSlotSyncResponse);
            remoteClusterSlotState.updateSlots(remoteClusterSlotSyncResponse.getSlots());
        }

        remoteClusterSlotState.updateDataCenterMetadata(remoteClusterSlotSyncResponse.getDataCenter());
    }

    public void handleFailResponse(RemoteClusterSlotSyncRequest remoteClusterSlotSyncRequest, GenericResponse<RemoteClusterSlotSyncResponse> genericResponse) {

        RemoteClusterSlotSyncResponse remoteClusterSlotSyncResponse = genericResponse.getData();

        if (remoteClusterSlotSyncResponse == null) {
            throw new RuntimeException(StringFormatter.format("sync request: {} fail, resp.data is null, msg: {}", remoteClusterSlotSyncRequest, genericResponse.getMessage()));
        }
        // heartbeat on follow, refresh leader;
        // it will sync on leader next time;
        if (!remoteClusterSlotSyncResponse.isSyncOnLeader()) {
            remoteClusterMetaExchanger.learn(remoteClusterSlotSyncRequest.getDataCenter(), new Leader(remoteClusterSlotSyncResponse.getMetaLeader(), remoteClusterSlotSyncResponse.getMetaLeaderEpoch()));
            // refresh the leader from follower, but the info maybe incorrect
            // throw the exception to trigger the counter inc
            // if the info is correct, the counter would be reset
            throw new RuntimeException(StringFormatter.format("sync dataCenter: {} on metaServer.follower, leader is: {} ", remoteClusterSlotSyncRequest.getDataCenter(), new Leader(remoteClusterSlotSyncResponse.getMetaLeader(), remoteClusterSlotSyncResponse.getMetaLeaderEpoch())));
        } else
            if (!remoteClusterSlotSyncResponse.isLeaderWarmuped()) {
                // remote leader not warmuped, just print log, not throw exception.
                logger.info("sync dataCenter: {}, remote leader:{} not warmuped.", remoteClusterSlotSyncRequest.getDataCenter());
            } else {
                throw new RuntimeException(StringFormatter.format("sync dataCenter: {} on metaServer.leader error, msg={}, data={}", genericResponse.getMessage(), remoteClusterSlotSyncResponse));
            }
    }

    public void handleSyncFail(RemoteClusterSlotSyncRequest remoteClusterSlotSyncRequest, Throwable t) {
        RemoteClusterSlotState remoteClusterSlotState = dataCenterToRemoteClusterSlotStateMap.get(remoteClusterSlotSyncRequest.getDataCenter());
        remoteClusterSlotState.incrementAndGetFailCount();
        logger.error("[syncRemoteMeta]sync request: {} error.", remoteClusterSlotSyncRequest, t);
    }

    public static final class RemoteClusterSlotState {
        final AtomicLong failCount = new AtomicLong(0);
        volatile Slots slots;
        volatile DataCenter dataCenter;
        volatile KeyedTask<RemoteClusterSlotSyncTask> task;

        public RemoteClusterSlotState() {
            this.slots = Slots.INIT;
        }

        public RemoteClusterSlotState(Slots slots, DataCenter dataCenter) {
            this.slots = slots;
            this.dataCenter = dataCenter;
        }

        public synchronized void updateSlots(Slots slots) {

            Slots prev = this.slots;
            if (this.slots.getEpoch() < slots.getEpoch()) {
                this.slots = slots;
                logger.info("Slots update from {} to {}, data: {}", prev.getEpoch(), slots.getEpoch(), slots);
            }
        }

        public synchronized void updateDataCenterMetadata(DataCenter dataCenter) {
            if (dataCenter != null && !dataCenter.equals(this.dataCenter)) {
                logger.info("dataCenterMetadata update from {} to {}", this.dataCenter, dataCenter);
                this.dataCenter = dataCenter;
            }
        }

        public long incrementAndGetFailCount() {
            return failCount.incrementAndGet();
        }

        public void initFailCount() {
            failCount.set(0);
        }

        public long getFailCount() {
            return failCount.get();
        }


        public synchronized DataCenter getDataCenterMetadata() {
            return dataCenter;
        }

        public synchronized Slots getSlots() {
            return this.slots;
        }


    }

}
